package com.fwmagic.spark.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object SocketStreaming {

  def main(args:Array[String]):Unit ={
    /***
      * 使用SparkContext创建RDD
      * SparkStreaming是对SparkContext的包装和增强，SparkStreaming也有一个抽象的数据集DSteam[离散数据流]，是对RDD的扩展和增强
      * 使用SparkStreaming创建DStream
      *
      */
    val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[*]")

    val sc = new SparkContext(conf)

    //设置日志级别
    sc.setLogLevel("WARN");

    val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))

    //Transformation
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost",8888)

    //Action
    val result: DStream[(String, Int)] = lines.flatMap(_.split("\\s")).map(word =>(word,1)).reduceByKey(_+_)

    result.print()

    //启动任务
    ssc.start()

    //等待任务停止
    ssc.awaitTermination()
  }
}
